package com.intct.cdc;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

/**
 * @author gufg
 * @since 2025-11-11 16:23
 */
public class CDCDataStreamAPI {
    public static void main(String[] args) throws Exception {
        // 获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置全局并行度
        env.setParallelism(1);

        // 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 配置源--mysqlcdc
        MySqlSource<String> sqlSource = MySqlSource.<String>builder()
                .hostname("cdh-node")
                .port(13306)
                .databaseList("m1")
                .tableList("m1.test01")
                .username("root")
                .password("Test_090110")
                /*
                    关于无主键表
                    从2.4.0 版本开始支持无主键表，使用无主键表必须设置 scan.incremental.snapshot.chunk.key-column，且只能选择非空类型的一个字段。
                    在使用无主键表时，需要注意以下两种情况。
                    配置 scan.incremental.snapshot.chunk.key-column 时，如果表中存在索引，请尽量使用索引中的列来加快 select 速度。
                    无主键表的处理语义由 scan.incremental.snapshot.chunk.key-column 指定的列的行为决定：
                    如果指定的列不存在更新操作，此时可以保证 Exactly once 语义。
                    如果指定的列存在更新操作，此时只能保证 At least once 语义。但可以结合下游，通过指定下游主键，结合幂等性操作来保证数据的正确性。
                 */
                .chunkKeyColumn(new ObjectPath("m1", "test06"), "id")
                .deserializer(new JsonDebeziumDeserializationSchema())
                // 解决mysql中decimal类型在flinkcdc输出乱码问题
                .debeziumProperties(getDebeziumProperties())
                /*
                启动模式
                    配置选项scan.startup.mode指定 MySQL CDC 使用者的启动模式。有效枚举包括：
                    initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。
                    earliest-offset：跳过快照阶段，从可读取的最早 binlog 位点开始读取
                    latest-offset：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
                    specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
                    timestamp：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。
                 */
                .startupOptions(StartupOptions.initial())
                .build();

        // 执行获取数据
        DataStreamSource<String> sourceStream = env.fromSource(sqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source");

        // 输出--print
//        sourceStream.print();

        // 输出--kafka
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "90000");
        //    配置kafka
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers("cdh-node:9092")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.builder()
                                .setTopic("intct86")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build())
                /*
                    KafkaSink 总共支持三种不同的语义保证（DeliveryGuarantee）。
                        对于 DeliveryGuarantee.AT_LEAST_ONCE 和 DeliveryGuarantee.EXACTLY_ONCE，Flink checkpoint 必须启用。
                        默认情况下 KafkaSink 使用 DeliveryGuarantee.NONE。 以下是对不同语义保证的解释：

                    DeliveryGuarantee.NONE 不提供任何保证：消息有可能会因 Kafka broker 的原因发生丢失或因 Flink 的故障发生重复。
                    DeliveryGuarantee.AT_LEAST_ONCE: sink 在 checkpoint 时会等待 Kafka 缓冲区中的数据全部被 Kafka producer 确认。
                             消息不会因 Kafka broker 端发生的事件而丢失，但可能会在 Flink 重启时重复，因为 Flink 会重新处理旧数据。
                    DeliveryGuarantee.EXACTLY_ONCE: 该模式下，Kafka sink 会将所有数据通过在 checkpoint 时提交的事务写入。
                            因此，如果 consumer 只读取已提交的数据（参见 Kafka consumer 配置 isolation.level），
                             在 Flink 发生重启时不会发生数据重复。然而这会使数据在 checkpoint 完成时才会可见，因此请按需调整 checkpoint 的间隔。
                            请确认事务 ID 的前缀（transactionIdPrefix）对不同的应用是唯一的，以保证不同作业的事务 不会互相影响！
                            此外，强烈建议将 Kafka 的事务超时时间调整至远大于 checkpoint 最大间隔 + 最大重启时间，否
                            则 Kafka 对未提交事务的过期处理会导致数据丢失。
                 */
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 事务ID前缀
                .setTransactionalIdPrefix("cdc-api-")
                // 建议将 Kafka 的事务超时时间调整至远大于 checkpoint 最大间隔 + 最大重启时间,
                // 否则 Kafka 对未提交事务的过期处理会导致数据丢失。
                .setKafkaProducerConfig(properties)
                .build();

        //    转出mysql --> kafka
        sourceStream.sinkTo(kafkaSink);

        // 启动作业
        env.execute("flink cdc intct86");
    }

    /**
     * Mysql与Flink类型映射
     * @return
     */
    private static Properties getDebeziumProperties(){
        Properties properties = new Properties();
//        properties.setProperty("converters", "dateConverters");
//        //根据类在那个包下面修改
//        properties.setProperty("dateConverters.type", "com.belle.dc.stram.MySqlDateTimeConverter");
//        properties.setProperty("dateConverters.format.date", "yyyy-MM-dd");
//        properties.setProperty("dateConverters.format.time", "HH:mm:ss");
//        properties.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
//        properties.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
//        properties.setProperty("dateConverters.format.timestamp.zone", "UTC+8");
        properties.setProperty("debezium.snapshot.locking.mode","none"); //全局读写锁，可能会影响在线业务，跳过锁设置
        properties.setProperty("include.schema.changes", "true");
        properties.setProperty("bigint.unsigned.handling.mode","long");
        properties.setProperty("decimal.handling.mode","double");
        return properties;
    }
}
